package com.intct.flink.window;

import com.intct.hbase.bean.WindowBean;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

/**
 * @author gufg
 * @since 2025-06-27 14:26
 */
public class CountWindowDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);

        // 从socket获取数据
        DataStreamSource<String> sourceDS = env.socketTextStream("jd-node", 8888);

        // Map转换(将字符转换为实体Bean)
        SingleOutputStreamOperator<WindowBean> mapDS = sourceDS.map(sources -> {
            String[] sourceList = sources.split(",");
            return WindowBean.builder()
                    .name(sourceList[0])
                    .ts(Long.valueOf(sourceList[1]))
                    .vc(Integer.valueOf(sourceList[2]))
                    .build();
        });

        // 按照指定Key分区
        KeyedStream<WindowBean, String> keyByDS = mapDS.keyBy(key -> key.getName());

        // 窗口-- 分配器(滚动窗口) 10秒一个窗口
        WindowedStream<WindowBean, String, GlobalWindow> countWindow = keyByDS.countWindow(3);
        SingleOutputStreamOperator<String> processDS = countWindow.process(new ProcessWindowFunction<WindowBean, String, String, GlobalWindow>() {
            @Override
            public void process(String key, ProcessWindowFunction<WindowBean, String, String, GlobalWindow>.Context context,
                                Iterable<WindowBean> elements, Collector<String> out) throws Exception {

                long count = elements.spliterator().estimateSize();

                long maxTimestamp = context.window().maxTimestamp();
                String maxTime = DateFormatUtils.format(maxTimestamp, "yyyy-MM-dd HH:mm:ss");
                out.collect("key=" + key + "窗口[时间:" + maxTime+"] ;数据的条数 = " + count + " 数据 = " + elements.toString());
            }
        });

        processDS.print();


        env.execute();
    }
}
